package com.bw.gmall.realtime.common.base;

import com.bw.gmall.realtime.common.Util.FlinkSourceUtil;
import com.bw.gmall.realtime.common.constant.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public abstract class BaseApp {



    //    没有具体实现
    //    用于后续调用实现
    public abstract void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) throws Exception;

    public  void start(int port, int Parallelism, String ckAndGroupId, String topic) throws Exception {

        //    1.创建流环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        System.setProperty("HADOOP_USER_NAME", "hadoop");

////      2.并行度
//        // 可以在代码中设置，在配置文件中，在参数中
        env.setParallelism(Parallelism);
////      3.状态后端及检查点相关配置
////      3.1 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
////      3.2 开启checkpoint
//        env.enableCheckpointing(5000);
////      3.3 设置checkpoint模式：精准一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
////      3.4 checkpoint储存
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://node00:Flin_gmall/stream" + ckAndGroupId);
////      3.5 checkpoint并发数
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
////      3.6 checkpoint 之间最小问题
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
////      3.7 checkpoint  的超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000);
////      3.8 job 取消时 checkpoint 保留策略
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


//      4. 消费kafka
        KafkaSource<String> kafkaSource = FlinkSourceUtil.getKafkaSource(ckAndGroupId, topic);

        DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");


        stream.print();

        handle(env,stream);


        env.execute();

    }




}
